package com.yun.util.kafka_es.kafkaTest;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.HashSet;
import java.util.Properties;
import java.util.Set;


public class KafkaListenerTest{

	private Properties consumer = new Properties();
	private String keystorePath = "D:\\publish-platform\\platform\\server.keystore.jks";
	private String truststorePath = "D:\\publish-platform\\platform\\server.truststore.jks";
	private KafkaConsumer<String, String> kafkaConsumer;

	public KafkaConsumer<String, String> getKafkaConsumer() {
		return kafkaConsumer;
	}

	private final static KafkaListenerTest listenerTest = new KafkaListenerTest();

	public static KafkaListenerTest getInstance(){
		return listenerTest;
	}

	public KafkaListenerTest() {
		
//		File keystorePathFile = new File(keystorePath);
//		File truststorePathFile = new File(truststorePath);

		// 去配置集群
		//consumer.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
		consumer.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.6:9092");
		// 指定消费者的组，只要group.id相同，就属于同一个消费者组
		consumer.put(ConsumerConfig.GROUP_ID_CONFIG, "singularity-framework");
		 // 重置offset
		consumer.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 指定自动提交偏移量（默认是false）
		consumer.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 每次自动提交偏移量的间隔时间
		consumer.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
		consumer.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		consumer.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		consumer.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "platform");// TODO
																		// 密码
		consumer.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "platform");// TODO 密码
		consumer.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "platform");// TODO
																			// 密码
		consumer.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "JKS");
		consumer.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "JKS");
		consumer.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS);
		//consumer.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
		consumer.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");// 将这个配置项加上,而且value值为空值,即可解决"不存在主题替代名称"这个问题
//		consumer.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keystorePathFile.getAbsolutePath());
		consumer.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keystorePath);
//		consumer.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststorePathFile.getAbsolutePath());
		consumer.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststorePath);
		kafkaConsumer = new KafkaConsumer<>(consumer);
	}

    /**
     * 设置要监听的topic
     */
	Set<TopicPartition> setTopic() {
	 	TopicPartition partition0 = new TopicPartition("Event", 0);
	 	TopicPartition partition1 = new TopicPartition("Guoneng-Agent", 0);
	 	TopicPartition partition2 = new TopicPartition("LA-LogTransfer", 0);
		/*List<String> topics = new ArrayList<>();
		topics.add("Event");
		topics.add("Guoneng-Agent");
		topics.add("LA-LogTransfer");*/
		 Set<TopicPartition> list = new HashSet<>();
		 list.add(partition0);
		 list.add(partition1);
		 list.add(partition2);
		 kafkaConsumer.assign(list);
		 return list;
	}

}
